package com.wudl.flink;

import com.mysql.jdbc.PreparedStatement;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;

/**
 * @ClassName : SinkToMySQL
 * @Description :
 * @Author :wudl
 * @Date: 2021-03-14 15:08
 */

public class SinkToMySQL extends RichSinkFunction<String> {

    PreparedStatement ps;
    private Connection connection;
    Statement statement;

    /**
     * open() 方法中建立连接，这样不用每次 invoke 的时候都要建立连接和释放连接
     *
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        connection = getConnection();
        statement = connection.createStatement();
//        ps = (PreparedStatement) this.connection.createStatement();
    }

    @Override
    public void close() throws Exception {
        super.close();
        //关闭连接和释放资源
        if (connection != null) {
            connection.close();
        }
        if (ps != null) {
            ps.close();
        }
        if (statement !=null)
        {
            statement.close();
        }
    }

    /**
     * 每条数据的插入都要调用一次 invoke() 方法
     *
     * @param value
     * @param context
     * @throws Exception
     */
    @Override
    public void invoke(String sql, Context context) {
        try {
            if (sql !=null) {
                statement.executeUpdate(sql);
            }
        } catch (Exception ex) {
            ex.printStackTrace();
            System.out.println("***************************");
            System.out.println(ex.getMessage());
        }
    }

    private static Connection getConnection() {
        Connection con = null;
        try {
            Class.forName("com.mysql.jdbc.Driver");
            con = DriverManager.getConnection("jdbc:mysql://192.168.1.153:3306/db_test?useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=UTC", "root", "slave012021@123ABM");
        } catch (Exception e) {
            System.out.println("-----------mysql get connection has exception , msg = " + e.getMessage());
        }
        return con;
    }
}
